package com.intellif.mozping.consumer;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

import java.util.List;

/**
 * 监听Mq消息
 *
 * @author admin
 * @version 1.0
 * @date 2018年10月23日
 * @see MqSinkReceiver
 * @since JDK1.8
 */
@Component
public class MqSinkReceiver {

    /**
     * @param consumerRecords 消息体
     * @param ack             ack
     */
    @KafkaListener(
            id = "singleConsumer",
            topics = "#{'${kafka.listener.single.partition.topic}'.split(',')}",
            containerFactory = "ackSingleContainerFactory")
    public void inputPersonfileNewCluster(List<ConsumerRecord> consumerRecords, Acknowledgment ack) {
        LogRecord.handle(consumerRecords, ack);
    }

    /**
     * @param consumerRecords 消息体
     * @param ack             ack
     */
    @KafkaListener(
            id = "concurrencyConsumer",
            topics = "#{'${kafka.listener.multiple.partition.topic}'.split(',')}",
            containerFactory = "ackConcurrencyContainerFactory")
    public void consumerListener(List<ConsumerRecord> consumerRecords, Acknowledgment ack) {
        LogRecord.handle(consumerRecords, ack);
    }


}